package com.nbcio.iot.queue;

import com.fasterxml.jackson.databind.ObjectMapper;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.redisson.api.RBlockingQueue;
import org.redisson.api.RDelayedQueue;
import org.redisson.api.RedissonClient;
import org.springframework.stereotype.Service;

import java.util.Map;
import java.util.concurrent.TimeUnit;

import com.nbcio.common.core.constant.GlobalConstants;

/**
 * IoT 消息队列服务
 * 基于 Redis 实现消息队列，解耦数据采集和规则执行
 *
 * @author nbacheng
 */
@Slf4j
@Service
@RequiredArgsConstructor
public class IotMessageQueueService {

    private final RedissonClient redissonClient;
    private final ObjectMapper objectMapper = new ObjectMapper();

    // 队列名称常量（使用全局前缀，避免租户上下文缺失时报错）
    public static final String DATA_COLLECTION_QUEUE = GlobalConstants.GLOBAL_REDIS_KEY + "iot:queue:data-collection";
    public static final String RULE_EXECUTION_QUEUE = GlobalConstants.GLOBAL_REDIS_KEY + "iot:queue:rule-execution";
    public static final String DEVICE_CONTROL_QUEUE = GlobalConstants.GLOBAL_REDIS_KEY + "iot:queue:device-control";

    /**
     * 发送数据采集消息到队列
     */
    public void sendDataCollectionMessage(Map<String, Object> message) {
        try {
            String messageJson = objectMapper.writeValueAsString(message);
            RBlockingQueue<String> queue = redissonClient.getBlockingQueue(DATA_COLLECTION_QUEUE);
            queue.offer(messageJson);
            log.info("[QUEUE] 数据采集消息已发送: {}", message.get("deviceCode"));
        } catch (Exception e) {
            log.error("[QUEUE] 发送数据采集消息失败: {}", e.getMessage());
        }
    }

    /**
     * 发送规则执行消息到队列
     */
    public void sendRuleExecutionMessage(Map<String, Object> message) {
        try {
            String messageJson = objectMapper.writeValueAsString(message);
            RBlockingQueue<String> queue = redissonClient.getBlockingQueue(RULE_EXECUTION_QUEUE);
            queue.offer(messageJson);
            log.info("[QUEUE] 规则执行消息已发送: {}", message.get("deviceCode"));
        } catch (Exception e) {
            log.error("[QUEUE] 发送规则执行消息失败: {}", e.getMessage());
        }
    }

    /**
     * 发送延迟消息
     */
    public void sendDelayedMessage(String queueName, Map<String, Object> message, long delay, TimeUnit timeUnit) {
        try {
            String messageJson = objectMapper.writeValueAsString(message);
            RBlockingQueue<String> queue = redissonClient.getBlockingQueue(queueName);
            RDelayedQueue<String> delayedQueue = redissonClient.getDelayedQueue(queue);
            delayedQueue.offer(messageJson, delay, timeUnit);
            log.info("[QUEUE] 延迟消息已发送: {} 延迟: {}{}", queueName, delay, timeUnit);
        } catch (Exception e) {
            log.error("[QUEUE] 发送延迟消息失败: {}", e.getMessage());
        }
    }

    /**
     * 消费数据采集消息
     */
    public String consumeDataCollectionMessage() {
        try {
            RBlockingQueue<String> queue = redissonClient.getBlockingQueue(DATA_COLLECTION_QUEUE);
            return queue.poll(5, TimeUnit.SECONDS); // 5秒超时
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
            log.debug("[QUEUE] 消费数据采集消息被中断");
            return null;
        } catch (Exception e) {
            // 检查是否是应用关闭导致的异常
            if (isApplicationShuttingDown(e)) {
                log.debug("[QUEUE] 应用正在关闭，停止消费消息");
                return null;
            }
            log.error("[QUEUE] 消费数据采集消息失败: {}", e.getMessage());
            return null;
        }
    }

    /**
     * 消费规则执行消息
     */
    public String consumeRuleExecutionMessage() {
        try {
            RBlockingQueue<String> queue = redissonClient.getBlockingQueue(RULE_EXECUTION_QUEUE);
            return queue.poll(5, TimeUnit.SECONDS); // 5秒超时
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
            log.debug("[QUEUE] 消费规则执行消息被中断");
            return null;
        } catch (Exception e) {
            // 检查是否是应用关闭导致的异常
            if (isApplicationShuttingDown(e)) {
                log.debug("[QUEUE] 应用正在关闭，停止消费消息");
                return null;
            }
            log.error("[QUEUE] 消费规则执行消息失败: {}", e.getMessage());
            return null;
        }
    }

    /**
     * 获取队列大小
     */
    public int getQueueSize(String queueName) {
        try {
            RBlockingQueue<String> queue = redissonClient.getBlockingQueue(queueName);
            return queue.size();
        } catch (Exception e) {
            log.error("[QUEUE] 获取队列大小失败: {}", e.getMessage());
            return -1;
        }
    }

    private boolean isApplicationShuttingDown(Exception e) {
        String msg = String.valueOf(e.getMessage());
        return msg.contains("unable to connect") || msg.contains("Connection closed") || msg.contains("shutdown");
    }
}
